

package org.example.flinksql.test;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.example.flinksql.test.custom.*;

/**
 * 注意事项：
 * 1.使用insert into t_dst select * from t_src时，两边字段顺序要一致
 * 2.FlinkSQL的关键字必须使用"`"标注，如字段scope是关键字，需要写为`scope`
 * 3.MySQL中字段为“bigint(N) unsigned”的字段在FlinkSQL中需要使用DECIMAL(N,0)替代。
 *   因为Flink SQL BIGINT类型默认转换为Java的Long类型，bigint(N) unsigned数值范围比Long大
 * 4.最好在定义FlinkSQL表时定义PRIMARY KEY，避免不知原因的主键冲突
 */
public class SyncAndTransEnterprise {

	public static void main(String[] args) throws Exception {
		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
		TableEnvironment tbEnv = TableEnvironment.create(bsSettings);

		String limit = "";
		if (null != args && 0 != args.length) {
			String limitStr = args[0];
			limit = "limit " + limitStr;
		}

		String srcHost = "localhost";
		String srcPort = "3306";
		String srcUsername = "root";
		String srcPassword = "root";
		String srcDb = "db_enterprise_0";
		String srcTb = "t_enterprise_0";
		String srcTbStruct = "eid STRING," +
				"id DECIMAL(20,0)," +
				"reg_no STRING," +
				"credit_no STRING," +
				"org_no STRING," +
				"name STRING," +
				"format_name STRING," +
				"category STRING," +
				"province STRING," +
				"econ_kind STRING," +
				"regist_capi STRING," +
				"actual_capi STRING," +
				"`scope` STRING," +
				"term_start STRING," +
				"term_end STRING," +
				"check_date STRING," +
				"`belong_org` STRING," +
				"oper_name STRING," +
				"oper_type STRING," +
				"oper_name_id STRING," +
				"start_date STRING," +
				"end_date STRING," +
				"status STRING," +
				"address STRING," +
				"type INT," +
				"type_desc STRING," +
				"title STRING," +
				"`longitude` DOUBLE," +
				"latitude DOUBLE," +
				"`gd_longitude` DOUBLE," +
				"gd_latitude DOUBLE," +
				"obj_id STRING," +
				"source STRING," +
				"tax_no STRING," +
				"collegues_num STRING," +
				"description STRING," +
				"created_time BIGINT," +
				"last_update_time BIGINT," +
				"logo_url STRING," +
				"status_code STRING," +
				"econ_type STRING," +
				"department STRING," +
				"url STRING," +
				"org_type STRING," +
				"row_update_time TIMESTAMP," +
				"province_code STRING," +
				"district_code STRING," +
				"title_code STRING," +
				"econ_kind_code STRING," +
				"regist_capi_new DECIMAL(40,6)," +
				"currency_unit STRING," +
				"`belong_org_code` STRING," +
				"credit_area_code STRING," +
				"revoke_reason STRING," +
				"revoke_date STRING," +
				"logout_date STRING," +
				"logout_reason STRING," +
				"group_name STRING," +
				"group_abbreviation STRING," +
				"revoked_certificates STRING," +
				"new_status_code STRING," +
				"type_new STRING," +
				"category_new STRING," +
				"PRIMARY KEY (eid) NOT ENFORCED";

		String dstHost = "localhost";
		String dstPort = "3306";
		String dstUsername = "flinksql";
		String dstPassword = "w0c53w0c53";
		String dstDb = "db_test";
		String dstTb = "t_enterprise";
		String dstTbStruct = "eid STRING," +
				"id DECIMAL(20,0)," +
				"enterprise_name STRING," +
				"standard_name STRING," +
				"group_name STRING," +
				"group_abbreviation STRING," +
				"logo_url STRING," +
				"oper_name STRING," +
				"oper_type STRING," +
				"oper_name_id STRING," +
				"title STRING," +
				"title_code INT," +
				"reg_capi DECIMAL(40,6)," +
				"rec_capi DECIMAL(40,6)," +
				"currency_unit STRING," +
				"business_url STRING," +
				"collegues_num STRING," +
				"belong_org STRING," +
				"department STRING," +
				"`scope` STRING," +
				"reg_no STRING," +
				"credit_no STRING," +
				"org_no STRING," +
				"econ_kind_code INT," +
				"type INT," +
				"type_new STRING," +
				"category_new STRING," +
				"status STRING," +
				"status_code INT," +
				"start_date DATE," +
				"term_start DATE," +
				"term_end DATE," +
				"check_date DATE," +
				"logout_date DATE," +
				"logout_reason STRING," +
				"revoke_date DATE," +
				"revoke_reason STRING," +
				"revoked_certificates STRING," +
				"province_code STRING," +
				"source STRING," +
				"created_time TIMESTAMP," +
				"row_update_time TIMESTAMP," +
				"PRIMARY KEY (eid) NOT ENFORCED";

		// src: use db_enterprise_0.t_enterprise_0
		tbEnv.executeSql(String.format("CREATE TABLE t_src (%s) WITH (" +
				" 'connector' = 'jdbc', " +
				" 'url' = 'jdbc:mysql://%s:%s/%s" +
				"?useSSL=false&useUnicode=true&characterEncoding=UTF-8', " +
				" 'table-name' = '%s', " +
				" 'username' = '%s', " +
				" 'password' = '%s' " +
				")", srcTbStruct, srcHost, srcPort, srcDb, srcTb, srcUsername, srcPassword));

		// dst: use db_test.t_enterprise
		tbEnv.executeSql(String.format("CREATE TABLE t_dst (%s) WITH (" +
				" 'connector' = 'jdbc', " +
				" 'url' = 'jdbc:mysql://%s:%s/%s?useSSL=false&useUnicode=true&characterEncoding=UTF-8', " +
				" 'table-name' = '%s', " +
				" 'username' = '%s', " +
				" 'password' = '%s' " +
				")", dstTbStruct, dstHost, dstPort, dstDb, dstTb, dstUsername, dstPassword));

		// dst: use print connector named t_out
//		tbEnv.executeSql(String.format("CREATE TABLE t_out (%s) WITH ('connector' = 'print')", dstTbStruct));


		tbEnv.createTemporaryFunction("transDate", TransDateFunc.class);
		tbEnv.createTemporaryFunction("transDatetime", TransDatetimeFunc.class);
		tbEnv.createTemporaryFunction("transInteger", TransIntegerFunc.class);
		tbEnv.createTemporaryFunction("TransTC", TransTotalCapiFunc.class);
		tbEnv.createTemporaryFunction("transString", TransStringFunc.class);

		Table table = tbEnv.sqlQuery(String.format("select eid,id," +
				"name as enterprise_name,format_name as standard_name,group_name,group_abbreviation," +
				"logo_url,oper_name,oper_type,oper_name_id,title, 1 as title_code," +
				"TransTotalCapiFunc(regist_capi_new) as reg_capi,TransTotalCapiFunc(actual_capi) as rec_capi," +
				"currency_unit,url as business_url,collegues_num,belong_org,department,`scope`,reg_no," +
				"credit_no,org_no,transInteger(econ_kind_code) as econ_kind_code,type,type_new," +
				"category_new,status,transInteger(new_status_code) as status_code," +
				"transDate(start_date) as start_date,transDate(term_start) as term_start," +
				"transDate(term_end) as term_end,transDate(check_date) as check_date," +
				"transDate(logout_date) as logout_date,transString(logout_reason) as logout_reason," +
				"transDate(revoke_date) as revoke_date," +
				"revoke_reason,revoked_certificates,province_code,source," +
				"transDatetime(created_time) as created_time,row_update_time" +
				" from t_src %s", limit));
		table.executeInsert("t_dst");
	}
}
